package com.hkbigdata.processapi;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author liuanbo
 * @creat 2024-05-14-19:08
 * @see 2194550857@qq.com
 */
public class Flink01_ProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("hadoop102", 9999)
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] split = value.split(",");
                        WaterSensor waterSensor = new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                        out.collect(waterSensor);
                    }
                });

        //1. 在processapi里面独有的定时器和侧输出流和状态编程
        //2.AbstractRichFunction 包含运行时上下文对象RuntimeContext() 和 open()生命周期方法
        //3.process api 定时器+侧输出流+状态编程+运行时上下文+生命周期方法

        //总结：1.子类包含父类的所有方法
        //2.父类是抽象类那么必须重写所有的抽象方法
        //3.如果说是接口的话，那么必须重新所有方法，无论是抽象还是非抽象方法
        //4.子类必须重写父类抽象方法，非抽象方法可以不用重写也可以重写。

        waterSensorSingleOutputStreamOperator.process(new ProcessFunction<WaterSensor, String>() {
            //表示处理数据的方法
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                ctx.timerService();
                ctx.output(new OutputTag<String>("id") {
                }, value.toString());
                ctx.timestamp();
                out.collect(value.toString());
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
            }

            @Override
            public void setRuntimeContext(RuntimeContext t) {
                super.setRuntimeContext(t);
            }

            @Override
            public RuntimeContext getRuntimeContext() {
                return super.getRuntimeContext();
            }

            @Override
            public IterationRuntimeContext getIterationRuntimeContext() {
                return super.getIterationRuntimeContext();
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
            }

            @Override
            public void close() throws Exception {
                super.close();
            }
        }).print();

        env.execute();

    }
}
